package com.roy.wordcount;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

public class SplitBolt extends BaseRichBolt {

    OutputCollector collector;

    //初始化方法
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.collector = outputCollector;
    }

    //执行方法，会有一个while(true)进行循环调用。tuple为上一个tuple或者spout发送过来的数据。
    public void execute(Tuple tuple) {
        //接收到WordCountSpout传过来的tuple。
        String line = tuple.getString(0);
//        String line = tuple.getStringByField("wordCount")
        String[] arrwords = line.split(" ");
        //模拟mapreduce的流程，发送一个word,1的元祖。
        for(String word:arrwords){
            collector.emit(new Values(word,1));
        }
    }
    //声明输入字段
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("word","sum"));
    }
}
